package monitor

import java.io.InputStream
import java.util
import java.util.Properties

import bean.BusInfo
import com.alibaba.druid.pool.DruidDataSourceFactory
import javax.sql.DataSource
import org.apache.commons.dbutils.QueryRunner
import org.apache.spark.sql.ForeachWriter

import scala.collection.JavaConverters._

object MySQLWriter {
  val p: Properties = new Properties()
  val inputStream: InputStream = this.getClass.getResourceAsStream("/druid.properties")
  p.load(inputStream)
  var dataSource: DataSource = DruidDataSourceFactory.createDataSource(p)

  def getDatasource: DataSource = {
    dataSource
  }
}

class MySQLWriter extends ForeachWriter[BusInfo]{
  var qr: QueryRunner = _
  override def open(partitionId: Long, epochId: Long): Boolean = {
    qr = new QueryRunner(MySQLWriter.getDatasource)
    true
  }

  override def process(value: BusInfo): Unit = {
    val sql = "insert into lg_bus_info(deployNum, lglat) values(?,?)"
    qr.update(sql, value.deployNum, value.lglat)
  }

  override def close(errorOrNull: Throwable): Unit = {

  }
}
